import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.log4j.BasicConfigurator;

public class RiskPredictionDriver {
	public static void main(String[] args) throws Exception {
		BasicConfigurator.configure();
		Configuration conf = new Configuration();
		Job job = Job.getInstance(conf);

		job.setJarByClass(RiskPredictionDriver.class);
		job.setMapperClass(RiskPredictionMapper.class);
		job.setReducerClass(RiskPredictionReducer.class);

		job.setMapOutputKeyClass(IotData.class);
		job.setMapOutputValueClass(NullWritable.class);
		job.setOutputKeyClass(IotData.class);
		job.setOutputValueClass(NullWritable.class);

		//添加自定义分区
		job.setPartitionerClass(RiskPredictionPartitioner.class);
		//设置Reducetasks个数
		job.setNumReduceTasks(5);

		Path in = new Path("src/BigData/datas/sqldatas.txt");
		Path out = new Path("src/BigData/result");

		FileInputFormat.addInputPath(job, in);
		FileOutputFormat.setOutputPath(job, out);

		boolean res = job.waitForCompletion(true);
		System.exit(res ? 0 : 1);


	}
}
